package com.example.canal;
 
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
 
/**
 * @Description: CanalTask启动类
 * @Author: leo
 * @Date: 2024/12/5
 */
@Slf4j
@Component
public class CanalRunClient implements ApplicationRunner {
 
    @Value("${canal.ip}")
    private String ip;
 
    @Value("${canal.port}")
    private Integer port;
 
    @Value("${canal.username}")
    private String username;
 
    @Value("${canal.password}")
    private String password;
 
    @Value("${canal.destination}")
    private String destination;
 
    @Value("${canal.batch-size}")
    private Integer batchSize;
 
    @Value("${canal.subscribe}")
    private String subscribe;
 
    @Autowired
    private MessageHandler messageHandler;
 
    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("----->>>>>>>>启动canal");
        startCanal();
    }
 
    private void startCanal() {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, port), destination, username, password);
        try {
            //打开连接
            connector.connect();
            //订阅数据库表，全部表
            connector.subscribe(subscribe);
//            connector.subscribe("seata_account\\..*");
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始
            connector.rollback();
            while (true) {
                //获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                //如果没有数据
                if (batchId == -1 || size == 0) {
                    try {
                        //现成休眠1s
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    //如果有数据，处理数据
                    messageHandler.handler(message);
                }
 
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
 
        } finally {
            connector.disconnect();
        }
    }
 
}